/*   1:    */ package com.taobao.txc.rm.mq;
/*   2:    */ 
/*   3:    */ import com.aliyun.openservices.ons.api.ONSFactory;
/*   4:    */ import com.aliyun.openservices.ons.api.Producer;
/*   5:    */ import com.aliyun.openservices.ons.api.impl.rocketmq.ProducerImpl;
/*   6:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.Validators;
/*   7:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.exception.MQClientException;
/*   8:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.impl.producer.DefaultMQProducerImpl;
/*   9:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.DefaultMQProducer;
/*  10:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.LocalTransactionState;
/*  11:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.SendResult;
/*  12:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.client.producer.SendStatus;
/*  13:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.Message;
/*  14:    */ import com.aliyun.openservices.shade.com.alibaba.rocketmq.common.message.MessageAccessor;
/*  15:    */ import com.taobao.txc.common.LoggerInit;
/*  16:    */ import com.taobao.txc.common.LoggerWrap;
/*  17:    */ import com.taobao.txc.resourcemanager.mt.MtResourceManager;
/*  18:    */ import java.util.Properties;
/*  19:    */ 
/*  20:    */ public class TxcMQProducerImpl
/*  21:    */   implements TxcMQProducer
/*  22:    */ {
/*  23: 26 */   private static final LoggerWrap logger = LoggerInit.logger;
/*  24:    */   private MtResourceManager resourceManager;
/*  25:    */   private DefaultMQProducer defaultMQProducer;
/*  26:    */   
/*  27:    */   public TxcMQProducerImpl(DefaultMQProducer defaultMQProducer)
/*  28:    */   {
/*  29: 33 */     this.defaultMQProducer = defaultMQProducer;
/*  30: 34 */     this.resourceManager = MtResourceManager.getTxcResourceManager();
/*  31:    */   }
/*  32:    */   
/*  33:    */   public TxcMQProducerImpl(String ProducerId, String AccessKey, String SecretKey)
/*  34:    */   {
/*  35: 38 */     Properties properties = new Properties();
/*  36: 39 */     properties.put("ProducerId", ProducerId);
/*  37: 40 */     properties.put("AccessKey", AccessKey);
/*  38: 41 */     properties.put("SecretKey", SecretKey);
/*  39:    */     
/*  40: 43 */     Producer producer = ONSFactory.createProducer(properties);
/*  41: 44 */     this.defaultMQProducer = ((ProducerImpl)producer).getDefaultMQProducer();
/*  42: 45 */     this.resourceManager = MtResourceManager.getTxcResourceManager();
/*  43:    */   }
/*  44:    */   
/*  45:    */   public SendResult send(String xid, long branchId, Message msg)
/*  46:    */   {
/*  47:    */     try
/*  48:    */     {
/*  49: 56 */       Validators.checkMessage(msg, this.defaultMQProducer);
/*  50:    */     }
/*  51:    */     catch (Exception e)
/*  52:    */     {
/*  53: 58 */       throw new RuntimeException("message validate error", e);
/*  54:    */     }
/*  55: 61 */     SendResult sendResult = null;
/*  56:    */     try
/*  57:    */     {
/*  58: 63 */       MessageAccessor.putProperty(msg, "TRAN_MSG", "true");
/*  59: 64 */       MessageAccessor.putProperty(msg, "PGROUP", this.defaultMQProducer.getProducerGroup());
/*  60: 65 */       sendResult = this.defaultMQProducer.send(msg);
/*  61:    */     }
/*  62:    */     catch (Exception e)
/*  63:    */     {
/*  64: 67 */       throw new RuntimeException("send message Exception", e);
/*  65:    */     }
/*  66: 70 */     SendStatus ss = sendResult.getSendStatus();
/*  67: 71 */     if ((ss == SendStatus.SEND_OK) && 
/*  68: 72 */       (sendResult.getTransactionId() != null)) {
/*  69: 73 */       MessageAccessor.putProperty(msg, "__transactionId__", sendResult.getTransactionId());
/*  70:    */     }
/*  71: 77 */     String udataOfSendResult = SendResult.encoderSendResultToJson(sendResult);
/*  72: 78 */     this.resourceManager.reportUdata(xid, branchId, "TxcMetaQProducer.send", udataOfSendResult, true);
/*  73: 79 */     return sendResult;
/*  74:    */   }
/*  75:    */   
/*  76:    */   public boolean rollback(String txId, long branchId, String udataOfSendResult)
/*  77:    */   {
/*  78: 84 */     SendResult sendResult = SendResult.decoderSendResultFromJson(udataOfSendResult);
/*  79: 85 */     LocalTransactionState localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
/*  80:    */     try
/*  81:    */     {
/*  82: 87 */       this.defaultMQProducer.getDefaultMQProducerImpl().endTransaction(sendResult, localTransactionState, null);
/*  83:    */     }
/*  84:    */     catch (Exception e)
/*  85:    */     {
/*  86: 89 */       logger.warn("local transaction execute " + localTransactionState + ",but end broker transaction failed,e:" + e.getMessage());
/*  87: 90 */       return false;
/*  88:    */     }
/*  89: 92 */     return true;
/*  90:    */   }
/*  91:    */   
/*  92:    */   public boolean commit(String txId, long branchId, String udataOfSendResult)
/*  93:    */   {
/*  94: 97 */     SendResult sendResult = SendResult.decoderSendResultFromJson(udataOfSendResult);
/*  95: 98 */     LocalTransactionState localTransactionState = LocalTransactionState.COMMIT_MESSAGE;
/*  96:    */     try
/*  97:    */     {
/*  98:100 */       this.defaultMQProducer.getDefaultMQProducerImpl().endTransaction(sendResult, localTransactionState, null);
/*  99:    */     }
/* 100:    */     catch (Exception e)
/* 101:    */     {
/* 102:102 */       logger.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed,e:" + e.getMessage());
/* 103:103 */       return false;
/* 104:    */     }
/* 105:105 */     return true;
/* 106:    */   }
/* 107:    */   
/* 108:    */   public void start()
/* 109:    */     throws MQClientException
/* 110:    */   {
/* 111:110 */     this.defaultMQProducer.start();
/* 112:    */   }
/* 113:    */   
/* 114:    */   public void shutdown()
/* 115:    */   {
/* 116:115 */     this.defaultMQProducer.shutdown();
/* 117:    */   }
/* 118:    */   
/* 119:    */   public MtResourceManager getResourceManager()
/* 120:    */   {
/* 121:119 */     return this.resourceManager;
/* 122:    */   }
/* 123:    */   
/* 124:    */   public void setResourceManager(MtResourceManager resourceManager)
/* 125:    */   {
/* 126:123 */     this.resourceManager = resourceManager;
/* 127:    */   }
/* 128:    */ }


/* Location:           C:\Users\ibm\Downloads\txc-yun-sample (1)\txc-yun-sample\lib\txc-client-2.0.72.jar
 * Qualified Name:     com.taobao.txc.rm.mq.TxcMQProducerImpl
 * JD-Core Version:    0.7.0.1
 */